package tom.spring.cloud.stream.sender;

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.support.GenericMessage;
import tom.spring.cloud.msg.MyInput;
import tom.spring.cloud.msg.MyOutput;

import java.util.Date;

/**
 * 此处EnableBinding注入了两个自定义管道接口MyInput/MyOutput。
 * spring-cloud-stream提供了Sink/Source，自定义接口就是参照这两个接口实现的。
 * 此外还提供了Processor继承了Sink和Source同时兼有输入和输出管道定义可以替代单独的两个类
 *
 * @author ZHUFEIFEI
 */
@Slf4j
@EnableBinding({MyOutput.class, MyInput.class})
public class Producer {

    /**
     * 此处生产者对应生产消息OUTPUT发送通道：input_1
     *       获取回馈信息(消费)的通道INPUT：output_1
     *
     *       消费者从input_1中获取消息，并回馈信息到output_1，
     *       生产者生产消息到input_1，并从output_1读取消息
     *
     *       这里虽然用了相同的类(MyInput/MyOutput)作为输入输出管道定义，但是与消费者对应的destination是相反的
     *
     *       见配置文件：application-producer.yml
     * @return
     */
    @Bean
    @InboundChannelAdapter(channel = MyOutput.OUTPUT, poller = {@Poller(fixedRate = "2000")})
    public MessageSource<Date> send010() {
        return () -> {Date d = new Date();log.info("send010 date: {}", d);return new GenericMessage<>(d);};
    }

    @StreamListener(MyInput.INPUT)
    public void feedback(String msg) {
        log.info("feedback : {}", msg);
    }
}
